package bigdata.backend.meta.utils;

import bigdata.backend.meta.dto.DorisTableStruct;

import java.util.ArrayList;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class DorisTableSqlParser {

    public static DorisTableStruct parse(String createSql) {
        DorisTableStruct struct = new DorisTableStruct();

        // 1. KEY 模型
        String[] patterns = {
                "(UNIQUE)\\s+KEY\\s*\\(([^)]+)\\)",
                "(DUPLICATE)\\s+KEY\\s*\\(([^)]+)\\)",
                "(AGGREGATE)\\s+KEY\\s*\\(([^)]+)\\)"
        };
        for (String pattern : patterns) {
            Matcher m = Pattern.compile(pattern, Pattern.CASE_INSENSITIVE).matcher(createSql);
            if (m.find()) {
                String keyType = m.group(1).toUpperCase();
                String colsStr = m.group(2);
                struct.setKeyType(keyType + "_KEY");
                struct.setKeyColumns(parseColumnList(colsStr));
                break;
            }
        }

        // 2. 分区字段
        Matcher m = Pattern.compile(
                "PARTITION\\s+BY\\s+RANGE\\s*\\(`([^`]+)`\\)",
                Pattern.CASE_INSENSITIVE
        ).matcher(createSql);
        if (m.find()) {
            struct.setPartitionColumn(m.group(1));
        }

        // 3. 分区值
        m = Pattern.compile(
                "PARTITION\\s+\\w+\\s+VALUES\\s+LESS\\s+THAN\\s*\\(\\s*\"([^\"]+)\"\\s*\\)",
                Pattern.CASE_INSENSITIVE
        ).matcher(createSql);
        while (m.find()) {
            struct.getPartitionValues().add(m.group(1));
        }

        // 4. 分桶
        m = Pattern.compile(
                "DISTRIBUTED\\s+BY\\s+HASH\\s*\\(([^)]+)\\)\\s+BUCKETS\\s+(\\d+)",
                Pattern.CASE_INSENSITIVE
        ).matcher(createSql);
        if (m.find()) {
            struct.setBucketColumns(parseColumnList(m.group(1)));
            struct.setBucketNum(Integer.parseInt(m.group(2)));
        }

        // 5. PROPERTIES
        m = Pattern.compile(
                "PROPERTIES\\s*\\(\\s*([^;]+?)\\s*\\);",
                Pattern.CASE_INSENSITIVE | Pattern.DOTALL
        ).matcher(createSql);
        if (m.find()) {
            String propsBlock = m.group(1);
            Matcher pairMatcher = Pattern.compile(
                    "\"([^\"]+)\"\\s*=\\s*\"([^\"]+)\"",
                    Pattern.CASE_INSENSITIVE
            ).matcher(propsBlock);
            while (pairMatcher.find()) {
                struct.getProperties().put(pairMatcher.group(1), pairMatcher.group(2));
            }
            if (struct.getProperties().containsKey("replication_num")) {
                try {
                    struct.setReplicationNum(Integer.parseInt(struct.getProperties().get("replication_num")));
                } catch (NumberFormatException ignored) {}
            }
        }

        return struct;
    }

    public static List<String> parseColumnList(String columnListStr) {
        List<String> cols = new ArrayList<>();
        if (columnListStr == null || columnListStr.trim().isEmpty()) {
            return cols;
        }
        String[] parts = columnListStr.split(",");
        for (String part : parts) {
            String col = part.trim();
            if (col.startsWith("`") && col.endsWith("`")) {
                col = col.substring(1, col.length() - 1);
            }
            if (!col.isEmpty()) {
                cols.add(col);
            }
        }
        return cols;
    }
}
